/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.io.IOException;
import java.util.Random;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.Options;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sss.client.DataPutter;
import org.sss.client.JobEngine;
import org.sss.client.SssClient;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.datatype.PackableInt;
import org.sss.mapreduce.datatype.PackableLong;

public class GenPoints {
  @SuppressWarnings("unused")
  private static Logger logger = LoggerFactory.getLogger(GenPoints.class);

  private static GroupID body(SssClient client, long numPoints) throws IOException, SssException {
    Random rnd = new Random(0);
    DataPutter<PackableLong, KMeansPoint> putter = 
      DataPutter.create(client, PackableLong.class, KMeansPoint.class);
    try {
      for (long i = 0; i < numPoints; i++) {
        putter.append(new PackableLong(i), 
            new KMeansPoint(rnd.nextDouble() * 100,
                            rnd.nextDouble() * 100));
      }

      return putter.getOutputGroupID();
    }
    finally {
      putter.close();
    }
  }

  public static class GenPointsMapper extends Mapper {
    public void map(Context context, PackableInt key, PackableLong value, Output<PackableLong, KMeansPoint> output) throws Exception {
      int idx = key.get();
      long num = value.get();
      Random rnd = new Random(idx);

      for (long i = 0;i < num;++i) {
        output.write(new PackableLong((i << 8) | idx), 
                     new KMeansPoint(rnd.nextDouble() * 100,
                                     rnd.nextDouble() * 100));
      }
    }
  }

  private static GroupID bodyParallely(SssClient client, long numPoints) throws IOException, SssException {

    GroupID seed = GroupID.createRandom(client);
    long nrec = ((long)client.getCluster().getStorageNodes().size()) * seed.getNSlot();
    long unit = (numPoints + nrec - 1)/nrec;

    DataPutter<PackableInt, PackableLong> putter = 
      DataPutter.create(client, PackableInt.class, PackableLong.class, seed);
    try {
      long rest = numPoints;
      for (int i = 0;i < client.getCluster().getStorageNodes().size();++i) {
        for (int j = 0;j < seed.getNSlot();++j) {
          PackableLong value = new PackableLong(Math.min(unit, rest));
          rest -= value.get();
          putter.append(new PackableInt((i << 4) | j), value);
        }
      }
      assert rest == 0;
    }
    finally {
      putter.close();
    }
    JobEngine engine = new JobEngine(client);
    try {
      GroupID points = GroupID.createRandom(engine);
      engine.getJobBuilder("GenPoints", GenPointsMapper.class)
        .addInput(seed).addOutput(points).build();
      engine.exec();
      return points;
    }
    finally {
      engine.close();
    }
  }

  /*
   * GenPoints NUM_POINTS
   */
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);

    Options options = new Options();
    options.addOption("p", "parallel", false, "run parallely.");

    CommandLineParser parser = new GnuParser();
    CommandLine cl = parser.parse(options, client.getArgs());

    args = cl.getArgs();
    if (args.length < 1) {
      System.err.println("usage: GenPoints NUM_POINTS");
      System.exit(1);
    }
    long numPoints = Long.parseLong(args[0]);

    GroupID points;
    if (cl.hasOption("p")) {
      points = bodyParallely(client, numPoints);
    }
    else {
      points = body(client, numPoints);
    }
    System.out.println("" + points);
  }
}
